package com.knight.hadoop.day12.zuoye;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 日志解析第三步 生成Visits表
 * 
 * @author
 *
 */
public class LogParseStepThree {

	static class LogParseStepThreeMapper extends Mapper<LongWritable, Text, LogBean2, LogBean2> {

		static LogBean2 kv = new LogBean2();

		@Override
		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
			String line = value.toString();
			String[] splitStr = line.split("\001");
			String sessionId = splitStr[0];
			String ip = splitStr[1];
			String remoteUser = splitStr[2];
			String date = splitStr[3];
			String requestURL = splitStr[4];
		//	String stepNum = splitStr[5];
	//		String stayTime = splitStr[6];
			String referal = splitStr[7];
			/*String userAgent = splitStr[8];
			String bodyBytesSend = splitStr[9];
			String status = splitStr[10];*/
			
			kv.set(date, ip, remoteUser, sessionId, requestURL, referal,"","","");
			context.write(kv, kv);
		}
	}

	static class LogParseStepThreeReducer extends Reducer<LogBean2, LogBean2, Text, NullWritable> {
		
		static NullWritable v = NullWritable.get();
		static Text k = new Text();
		
		@Override
		protected void reduce(LogBean2 key, Iterable<LogBean2> values, Context context)
				throws IOException, InterruptedException {
			List<LogBean2> listBean = new ArrayList<>();

			for (LogBean2 bean : values) {
				LogBean2 logBean2 = new LogBean2();
				try {
					BeanUtils.copyProperties(logBean2, bean);
				} catch (IllegalAccessException e) {
					e.printStackTrace();
				} catch (InvocationTargetException e) {
					e.printStackTrace();
				}
				listBean.add(logBean2);
			}
			
			int size = listBean.size();
			LogBean2 firstBean = listBean.get(0);
			LogBean2 endBean = listBean.get(size-1);
			
			
			StringBuilder sb = new StringBuilder();
			sb.append(firstBean.getSessionId()+"\001");//sessionId
			sb.append(firstBean.getIp()+"\001");//ip地址
			sb.append(firstBean.getDate()+"\001");//进入时间
			sb.append(endBean.getDate()+"\001");//离开时间
			sb.append(firstBean.getRequestURL()+"\001");//进入页面
			sb.append(endBean.getRequestURL()+"\001");//离开页面
			sb.append(firstBean.getReferal()+"\001");//来源地址
			sb.append(size);//访问页面数
			//sb.append(firstBean.getRemoteUser()+"\001");//用户
			
			k.set(sb.toString());
			
			context.write(k, v);
			
		}
	}
	
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);
		job.setJarByClass(LogParseStepTwo.class);
		job.setMapperClass(LogParseStepThreeMapper.class);
		job.setReducerClass(LogParseStepThreeReducer.class);
		job.setGroupingComparatorClass(GroupLogSessionComparator.class);

		job.setMapOutputKeyClass(LogBean2.class);
		job.setMapOutputValueClass(LogBean2.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);

		//FileInputFormat.setInputPaths(job, new Path("C:\\application\\book\\Hadoop\\output\\weblog\\step2"));
		//FileOutputFormat.setOutputPath(job, new Path("C:\\application\\book\\Hadoop\\output\\weblog\\step3"));
		FileInputFormat.setInputPaths(job, new Path("E:\\home\\webLog\\output2"));
		FileOutputFormat.setOutputPath(job, new Path("E:\\home\\webLog\\output3"));

		boolean res = job.waitForCompletion(true);

		System.exit(res ? 0 : 1);
	}
}
